package com.zlm.app;

import com.zlm.bean.PageViewCount;
import com.zlm.bean.UserBehavior;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.util.Random;

/**
 * Author: Harbour
 * Date: 2021-05-16 14:45
 * Desc: page view count 统计
 */
public class PageViewCountApp {
    public static void main(String[] args) throws Exception {
        // step 1 搭建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // step 2 获取输入数据流
        URL resource = PageViewCountApp.class.getResource("/UserBehavior.csv");
        System.out.println(resource.getPath());
        DataStream<String> inputDataStream = env.readTextFile(resource.getPath());

        // step 3 转换数据并声明watermark
        DataStream<UserBehavior> userBehaviorDataStream = inputDataStream.filter(data -> "pv".equalsIgnoreCase(data.split(",")[3])).map(data -> {
            String[] fields = data.split(",");
            int random = new Random().nextInt() * 10;
            return new UserBehavior(
                    new Long(fields[0]),
                    new Long(fields[1]),
                    new Integer(fields[2]),
                    fields[3],
                    new Long(fields[4])
            );
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
            @Override
            public long extractAscendingTimestamp(UserBehavior element) {
                return element.getTimestamp() * 1000L;
            }
        });

        // step 4 处理数据，开窗 (为了防止数据倾斜，使用随机数生成key将数据进行分组，再使用滚动窗口聚合)
        DataStream<PageViewCount> aggDataStream = userBehaviorDataStream.map(new MapFunction<UserBehavior, Tuple2<Integer, Long>>() {
            @Override
            public Tuple2<Integer, Long> map(UserBehavior value) throws Exception {
                return new Tuple2<Integer, Long>(new Random().nextInt() * 10, 1L);
            }
        })
        .keyBy(data -> data.f0)
        .timeWindow(Time.hours(1))
        .aggregate(new MyPVAggregateFunction(), new MyPVWindowFunction());

        aggDataStream.keyBy(PageViewCount::getWindowEnd)
            .process(new MyPVCountProcessFunction())
            .print();

        env.execute("pv count");
    }

    /**
     * 聚合page view
     */
    private static class MyPVAggregateFunction implements AggregateFunction<Tuple2<Integer, Long>, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(Tuple2<Integer, Long> value, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    /**
     * 实现窗口全局聚合
     * 输入类型是Long ，输出类型转为pv count， 聚合的key是integer随机数，窗口类型是时间窗口
     * PageViewCount 参数为：随机数转换的字符串， 窗口的结束时间戳， 前一阶段聚合的count
     */
    private static class MyPVWindowFunction implements WindowFunction<Long, PageViewCount, Integer, TimeWindow> {

        @Override
        public void apply(Integer integer, TimeWindow window, Iterable<Long> input, Collector<PageViewCount> out) throws Exception {
            out.collect(new PageViewCount(integer.toString(), window.getEnd(), input.iterator().next()));
        }
    }

    private static class MyPVCountProcessFunction extends KeyedProcessFunction<Long, PageViewCount, String> {

        // 保存page view count聚合结果的state
        ValueState<Long> pageViewCountState;

        @Override
        public void open(Configuration parameters) throws Exception {
            pageViewCountState = getRuntimeContext().getState(new ValueStateDescriptor<>("total-pv-count", Long.class));
        }

        @Override
        public void processElement(PageViewCount value, Context ctx, Collector<String> out) throws Exception {
            if (pageViewCountState.value() != null) {
                pageViewCountState.update(value.getCount() + pageViewCountState.value());
            } else {
                pageViewCountState.update(value.getCount());
            }
            ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect(">>>>>>>>>" + ctx.getCurrentKey() + ">>>>>>>>>" + pageViewCountState.value());
            pageViewCountState.clear();
        }
    }
}
